#include "xlcomm/net/tcp_server.h"

#include "xlcomm/base/logger.h"
#include "xlcomm/net/detail/acceptor.h"
#include "xlcomm/net/detail/event_loop_thread_pool.h"
#include "xlcomm/net/detail/sockets_ops.h"

namespace xlcomm {
namespace net {

void DefaultServerConnection(const TcpConnectionPtr& conn) {
  LOG_INFO << __FUNCTION__ << " " << conn->name() << " is "
           << (conn->IsConnected() ? "UP" : "DOWN")
           << ", local:" << conn->local_addr().ToIpPort();
}

void DefaultServerMessage(const TcpConnectionPtr& conn, Buffer* buf,
                          Timestamp) {
  LOG_INFO << __FUNCTION__ << " " << conn->name() << " only retrieve all";
  buf->HasReadAll();
}

TcpServer::TcpServer(EventLoop* loop, const InetAddress& listen_addr,
                     const std::string& name, Option option)
    : loop_(loop),
      ip_port_(listen_addr.ToIpPort()),
      name_(name),
      conn_id_(1),
      acceptor_(
          std::make_unique<Acceptor>(loop, listen_addr, option == kReusePort)),
      thread_pool_(std::make_shared<EventLoopThreadPool>(loop_, name)),
      started_(0),
      connect_callback_(DefaultServerConnection),
      disconnect_callback_(DefaultServerConnection),
      message_callback_(DefaultServerMessage) {
  LOG_INFO << "TcpServer [" << name_ << "] create";
  XLCOMM_ASSERT(loop_ != nullptr);
  acceptor_->SetNewConnectionCallback(
      std::bind(&TcpServer::NewConnection, this, _1, _2));
}

TcpServer::~TcpServer() {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  LOG_INFO << "TcpServer [" << name_ << "] dtor";

  for (auto& item : connections_) {
    TcpConnectionPtr conn(item.second);
    item.second.reset();
    EventLoop* io_loop = conn->loop();
    io_loop->RunInLoop(std::bind(&TcpConnection::ConnectDestroyed, conn));
  }
}

void TcpServer::Start(int thread_num, const ThreadInitCallback& cb) {
  if (++started_ == 1) {
    LOG_INFO << "TcpServer [" << name_ << "] start...";
    XLCOMM_ASSERT(thread_num >= 0);
    XLCOMM_ASSERT(!acceptor_->listening());
    thread_pool_->Start(thread_num, cb);
    loop_->RunInLoop(std::bind(&Acceptor::Listen, acceptor_.get()));
  }
}

void TcpServer::NewConnection(int sockfd, const InetAddress& peer_addr) {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  char buff[64];
  snprintf(buff, sizeof(buff), "#%lu-%s", conn_id_,
           peer_addr.ToIpPort().c_str());
  ++conn_id_;
  std::string conn_name = "'" + name_ + buff + "'";
  InetAddress local_addr(sockets::GetLocalAddr(sockfd));
  LOG_INFO << "TcpServer [" << name_ << "] accept new connection " << conn_name
           << ", local:" << local_addr.ToIpPort() << " conn_fd=" << sockfd;

  EventLoop* io_loop = thread_pool_->GetNextLoop();
  // FIXME: poll with zero timeout to double confirm the new connection
  TcpConnectionPtr conn = std::make_shared<TcpConnection>(
      io_loop, conn_name, sockfd, local_addr, peer_addr);
  connections_[conn_name] = conn;
  conn->set_connect_callback(connect_callback_);
  conn->set_disconnect_callback(disconnect_callback_);
  conn->set_message_callback(message_callback_);
  conn->set_write_complete_callback(write_complete_callback_);
  conn->set_close_callback(std::bind(&TcpServer::RemoveConnection, this,
                                     _1));  // FIXME: unsafe
  io_loop->RunInLoop(std::bind(&TcpConnection::ConnectEstablished, conn));
  // 在没有用mutex的情况下，借助RunInLoop实现多线程TcpServer
  // loop_和io_loop间的线程切换只发生在连接建立和销毁时，不影响正常业务的性能
}

void TcpServer::RemoveConnection(const TcpConnectionPtr& conn) {
  loop_->RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop, this,
                             conn));  // FIXME: unsafe
}

void TcpServer::RemoveConnectionInLoop(const TcpConnectionPtr& conn) {
  XLCOMM_ASSERT(loop_->IsInLoopThread());
  LOG_INFO << "closed, TcpServer [" << name_ << "] remove connection "
           << conn->name();
  XLCOMM_ASSERT(connections_.erase(conn->name()) == 1);
  EventLoop* io_loop = conn->loop();
  // 让conn的生命期长到调用ConnectDestroyed的时刻
  io_loop->QueueInLoop(std::bind(&TcpConnection::ConnectDestroyed, conn));
}

}  // namespace net
}  // namespace xlcomm